package com.atguigu.orderpay_detect

import com.atguigu.orderpay_detect.TxMacthDetect.getClass
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @Author: ys
 * @Description: 通过使用join方式合并两条流，但是只能处理匹配事件，不能处理不匹配事件(侧输出流)
 * @Date: create in 2020/5/20 10:16
 */
object TxMatchByJoin {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取订单事件流
    val resource = getClass.getResource("/OrderLog.csv")
    val orderEventStream = environment.readTextFile(resource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
      })
      .filter(_.txId != "")
      //设置时间戳
      //方式一(简单实现)：升序数据，不设置延时 到点就发车，从字段中抽取
      .assignAscendingTimestamps(_.eventTime * 1000L)
      .keyBy(_.txId)

    //读取支付到账事件流
    val receiptResource = getClass.getResource("/ReceiptLog.csv")
    val receiptEventStream = environment.readTextFile(receiptResource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        ReceiptEvent(dataArray(0).trim, dataArray(1).trim, dataArray(2).trim.toLong)
      })
      .assignAscendingTimestamps(_.eventTime * 1000L)
      //和前面orderEventStream处理保持一致，
      .keyBy(_.txId)

    //join处理: 间隔join 根据时间区间合并流
    val processedFunction = orderEventStream.intervalJoin(receiptEventStream)
      .between(Time.seconds(-5), Time.seconds(5))
      // 不能输出侧输出流了，只能处理匹配事件
      .process(new TxMatchByJoin())

    processedFunction.print("")

    environment.execute("tx pay match by join.")
  }
}

class TxMatchByJoin extends ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]{
  override def processElement(left: OrderEvent, right: ReceiptEvent, ctx: ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    out.collect(left, right)
  }
}
